package com.learning.crpc.consummer;

import com.learning.crpc.model.CrpcRequest;
import com.learning.crpc.model.CrpcResponse;
import io.netty.channel.Channel;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;

/**
 * @program: diy_rpc
 * @description:
 * @author: CaoYong
 * @create: 2021-07-01 17:47
 **/
public class CrpcConnection {
    final static ConcurrentHashMap<String, CompletableFuture> invokeMap = new ConcurrentHashMap<>();
    Channel channel;

    public CrpcConnection(Channel channel) {
        this.channel = channel;
    }

    public Object call(CrpcRequest crpcRequest) throws ExecutionException, InterruptedException {
        CompletableFuture future = new CompletableFuture();
        invokeMap.putIfAbsent(crpcRequest.getId(), future);
        channel.writeAndFlush(crpcRequest);
        return future.get();
    }

    public void complete(CrpcResponse crpcResponse) {
        CompletableFuture future = invokeMap.get(crpcResponse.getRequestId());
        future.complete(crpcResponse.getContent());
        invokeMap.remove(crpcResponse.getRequestId());
    }


}
